package com.loginfail_detect

import java.util

import com.loginfail_detect.bean.{LoginEvent, LoginFailWarning}
import org.apache.flink.cep.PatternSelectFunction
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/12/4 20:37
  *                 统计2秒钟连续登录失败，并输出
  *                 CEP操作,匹配模式类似于正则表达式
  */
object LoginFailWithCep {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //定义事件时间语义

    //从文件中读取数据，并转换成样例类,提取时间戳生成watermark
    val filePath: String = "LoginFailDetect/src/main/resources/LoginLog.csv"
    val fileStream: DataStream[String] = env.readTextFile(filePath)

    //转换成样例类类型，并指定时间戳和watermark
    val loginEventStream: DataStream[LoginEvent] = fileStream.map(data => {
      val arr: Array[String] = data.split(",")
      LoginEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
    })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(3)) {
        override def extractTimestamp(t: LoginEvent): Long = t.timestamp * 1000L
      })

    //定义匹配的模式，要求是一个登录失败事件后，紧跟另一个失败事件(2s内2次登录失败)
    val loginFailPattren: Pattern[LoginEvent, LoginEvent] = Pattern.begin[LoginEvent]("firstFail").where(_.eventType == "fail")
      .next("secondFail").where(_.eventType == "fail")
      .within(Time.seconds(2))

    Pattern.begin[LoginEvent]("firstFail").where(_.eventType == "fail")
      .next("secondFail").where(_.eventType == "fail")
      .next("thirdFail").where(_.eventType == "fail")
      .within(Time.seconds(5))

    //将模式应用到数据流上，得到一个PatternStream
    val patternStream: PatternStream[LoginEvent] = CEP
      .pattern(loginEventStream.keyBy(_.userId), loginFailPattren)

    //检出符合模式的数据流，需要调用select
    val loginFailWarningStream: DataStream[LoginFailWarning] = patternStream
      .select(new LoginFailEventMatch)

    loginFailWarningStream.print("fail")

    env.execute("login fail with cep job")

  }

  //实现自定义PattrenSelectFunction
  class LoginFailEventMatch extends PatternSelectFunction[LoginEvent, LoginFailWarning] {
    override def select(map: util.Map[String, util.List[LoginEvent]]): LoginFailWarning = {
      //当前匹配到的时间序列，就保存到map里
      val firstFailEvent = map.get("firstFail").get(0)
      val secondFailEvent = map.get("secondFail").iterator().next()
      LoginFailWarning(
        firstFailEvent.userId,
        firstFailEvent.timestamp,
        secondFailEvent.timestamp,
        "login fail")

    }
  }


}
